package tsif.tcluster.netty.impl;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

/**
 * Netty管理器(管理多个客户端连接)
 * 
 */
public abstract class NettyClientMgr {
	protected final NettyServer.EventLoopGroups groups = new NettyServer.EventLoopGroups();
	protected Bootstrap bootstrap;

	public synchronized boolean start(int threads) {
		EventLoopGroup workerGroup = createChildGroup("NettyClientMgr", threads);
		bootstrap = new Bootstrap();
		bootstrap.group(workerGroup);
		bootstrap.channel(NioSocketChannel.class);
		bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
		bootstrap.option(ChannelOption.TCP_NODELAY, true);
		bootstrap.option(ChannelOption.SO_RCVBUF, 1024 * 32);
		bootstrap.option(ChannelOption.SO_SNDBUF, 1024 * 32);
		bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000);
		bootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT); // 使用内存池

		if (!initOption(bootstrap)) {
			throw new RuntimeException("init netty error! threads=" + threads);
		}
		bootstrap.handler(createHandler());
		return true;
	}

	public synchronized void stop() {
		if (bootstrap == null) {
			return;
		}
		bootstrap = null;

		groups.shutdown();
	}

	public Bootstrap getBootstrap() {
		return bootstrap;
	}

	/** 初始化参数 **/
	protected abstract boolean initOption(Bootstrap bootstrap);

	/** 创建处理接口 **/
	protected abstract ChannelHandler createHandler();

	/** 这个是用于处理accept到的channel的eventloop **/
	protected NioEventLoopGroup createChildGroup(String name, int thread) {
		return groups.createGroup(name, thread);
	}

	/** 连接服务器 **/
	public Channel connect(String host, int port) {
		try {
			// connect
			ChannelFuture future = bootstrap.connect(host, port);
			// sync to wait
			future.syncUninterruptibly();
			// check
			if (future.isSuccess() && future.channel().isActive()) {
				return future.channel();
			} else {
				throw new RuntimeException("connect fail! " + port);
			}
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}

}
